package com.wmx.rocketmq.helloFriend;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.Date;
import java.util.Map;

/**
 * 生产者 - 发送消息
 * 1、RocketMQTemplate 默认使用 UTF-8 编码，可以通过{@link RocketMQTemplate#setCharset(java.lang.String)} 修改
 * 2、{@link RocketMQTemplate#getProducer()} 可以获取原生的 {@link DefaultMQProducer} 对象
 *
 * @author wangMaoXiong
 * @version 1.0
 * @date 2020/11/8 16:41
 */
@RestController
public class MyProducer {

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 同步发送消息
     * http://localhost:8080/helloFriend/synchronouslySendMsg?topic=WMX_TOP
     *
     * @param topic   :主题名称, 如 WMX_TOP,或者 WMX_TOP:TagA
     * @param dataMap : 消息对象,如 {"id":"787878TYTY","price":78787.878,"summary":"你好！RocketMQ convertAndSend"}
     * @return
     */
    @PostMapping("helloFriend/synchronouslySendMsg")
    public String synchronouslySendMsg(@RequestParam String topic, @RequestBody Map<String, Object> dataMap) {
        /**
         * void convertAndSend(D destination, Object payload): 将消息对象先进行序列化转换, 然后发送到给定的主题
         * destination: 消息发送到的目的地,不存在时自动创建, 格式为: topicName:tags(主题名:标签名)
         * payload: 待发送的消息对象,会自动序列化{@link MessageConverter}为 {@link Message} 进行发送, 无论是普通的 String,或者 POJO 对象都可以
         *<p>
         * 其中还重载了许多 convertAndSend 方法, 这些都属于封装程度较高的方法,底层也是调用的 send, doSend 方法
         */
        try {
            rocketMQTemplate.convertAndSend(topic, dataMap);
        } catch (MessagingException e) {
            ////消息发送失败，需要有补偿机制，比如临时保存到数据库中，后期再发送
            e.printStackTrace();
        }

        /**
         *销毁rocketMQTemplate,注意一旦销毁,则无法再继续使用
         * <code>rocketMQTemplate.destroy();</code>
         */
        return "Success! " + new Date();
    }

    /**
     * 向指定的目的地发送消息。rocketMQTemplate 底层也是同步发送：
     * {@link RocketMQTemplate#syncSend(java.lang.String, org.springframework.messaging.Message)}
     * http://localhost:8080/helloFriend/sendMsgObj?topic=WMX_TOP
     *
     * @param topic   :主题名称, 如 WMX_TOP, 或者 WMX_TOP:TagA
     * @param dataMap : 消息对象,如 {"id":"787878TYFGH","price":178787.878,"summary":"你好！RocketMQ send."}
     * @return
     */
    @PostMapping("helloFriend/sendMsgObj")
    public String sendMsg(@RequestParam String topic, @RequestBody Map<String, Object> dataMap) {
        /**
         * send(D destination, Message<?> message) : 向指定目的地发送消息. convertAndSend 底层也是调用此方法
         * destination: 消息发送到的目的地,不存在时自动创建, 格式为: topicName:tags(主题名:标签名)
         * message: 消息对象
         * <p></p>
         * MessageBuilder#withPayload(java.lang.Object) :消息构建器,将对象构建成消息,无论是普通的 String,或者 POJO 对象都可以
         */
        try {
            rocketMQTemplate.send(topic, MessageBuilder.withPayload(dataMap).build());
        } catch (Exception e) {
            //消息发送失败，需要有补偿机制，比如临时保存到数据库中，后期再发送
            e.printStackTrace();
        }
        return "Success! " + new Date();
    }

    /**
     * 同步发送消息
     * http://localhost:8080/helloFriend/syncSend?topic=WMX_TOP
     *
     * @param topic   :主题名称, 如 WMX_TOP, 或者 WMX_TOP:TagA
     * @param dataMap : 消息对象,如 {"id":121232,"price":545345.878,"summary":"你好噢！RocketMQ syncSend."}
     * @return
     */
    @PostMapping("helloFriend/syncSend")
    public String syncSend(@RequestParam String topic, @RequestBody Map<String, Object> dataMap) {
        /**
         * SendResult syncSend(String destination, Message<?> message) : 向指定目的地发送消息
         * SendResult syncSend(String destination, Message<?> message, long timeout)
         * SendResult syncSend(String destination, Collection<T> messages)
         * SendResult syncSend(String destination, Collection<T> messages, long timeout)
         * SendResult syncSend(String destination, Message<?> message, long timeout, int delayLevel)
         * SendResult syncSend(String destination, Object payload)
         * destination: 消息发送到的目的地,不存在时自动创建, 格式为: topicName:tags(主题名:标签名)
         * message: 消息对象
         * messages：消息对象集合，一次性发送多条
         * timeout；发送超时时间（毫秒）
         * delayLevel：延迟消息的级别，值为 [1,18] 分别对应 "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
         * <p></p>
         * MessageBuilder#withPayload(java.lang.Object) :消息构建器,将对象构建成消息,无论是普通的 String,或者 POJO 对象都可以
         */
        try {
            SendResult sendResult = rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(dataMap).build());
            System.out.println("发送结果：" + sendResult);
        } catch (Exception e) {
            //消息发送失败，需要有补偿机制，比如临时保存到数据库中，后期再发送
            e.printStackTrace();
        }
        return "Success! " + new Date();
    }

    /**
     * 异步发送消息
     * http://localhost:8080/helloFriend/asyncSend?topic=WMX_TOP:cat
     *
     * @param topic   :主题名称, 如 WMX_TOP, 或者 WMX_TOP:TagA
     * @param dataMap : 消息对象,如 {"id":"787878TYFGH","price":178787.878,"summary":"你好！RocketMQ asyncSend."}
     * @return
     */
    @PostMapping("helloFriend/asyncSend")
    public String asyncSend(@RequestParam String topic, @RequestBody Map<String, Object> dataMap) {
        /**
         * asyncSend(String destination, Object payload, SendCallback sendCallback): 异步发送消息
         * asyncSend(String destination, Message<?> message, SendCallback sendCallback)
         * asyncSend(String destination, Object payload, SendCallback sendCallback, long timeout)
         * asyncSend(String destination, Message<?> message, SendCallback sendCallback, long timeout)
         * asyncSend(String destination, Message<?> message, SendCallback sendCallback, long timeout, int delayLevel)
         * destination: 消息发送到的目的地,不存在时自动创建, 格式为: topicName:tags(主题名:标签名)
         * payload: 待发送的消息对象,会自动序列化{@link MessageConverter}为 {@link Message} 进行发送, 无论是普通的 String,或者 POJO 对象都可以
         * sendCallback: 消息发送结果回调, 发送成功或者失败进入不同的方法
         * message：消息对象
         * timeout；发送超时时间（毫秒）
         * delayLevel：延迟消息的级别，值为 [1,18] 分别对应 "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"
         * payload 转 message： Message<?> message = MessageBuilder.withPayload(payload).build()
         */
        rocketMQTemplate.asyncSend(topic, dataMap, new SendCallback() {
            @Override
            public void onSuccess(SendResult var1) {
                System.out.printf("异步发送成功, SendResult=%s %n", var1);
            }

            @Override
            public void onException(Throwable var1) {
                //消息发送失败，需要有补偿机制，比如临时保存到数据库中，后期再发送
                System.out.printf("异步发送失败, Throwable=%s %n", var1);
            }
        });
        return "Success! " + new Date();
    }


    /**
     * 同步发送顺序消息
     * http://localhost:8080/helloFriend/syncSendOrderly?topic=WMX_TOP:cat
     * 顺序发送消息,消息有序指的是可以按照消息的发送顺序来消费(FIFO)。RocketMQ可以严格的保证消息有序，可以分为分区有序或者全局有序。
     * 以订单进行分区有序的示例:
     * 一个订单的顺序流程是：创建、付款、推送、完成。订单号相同的消息会被先后发送到同一个队列中，消费时，同一个OrderId获取到的肯定是同一个队列。
     * https://github.com/apache/rocketmq/blob/master/docs/cn/RocketMQ_Example.md
     *
     * @param topic   :主题名称, 如 WMX_TOP, 或者 WMX_TOP:TagA
     * @param dataMap : 消息对象,如 {"id":"787878TYFGH","orderId":"p9009","price":178787.878,"summary":"你好！RocketMQ syncSendOrderly."}
     * @return
     */
    @PostMapping("helloFriend/syncSendOrderly")
    public String syncSendOrderly(@RequestParam String topic, @RequestBody Map<String, Object> dataMap) {
        /**
         * SendResult syncSendOrderly(String destination, Message<?> message, String hashKey): 同步发送顺序消息
         * SendResult syncSendOrderly(String destination, Message<?> message, String hashKey, long timeout)
         * SendResult syncSendOrderly(String destination, Object payload, String hashKey)
         * SendResult syncSendOrderly(String destination, Object payload, String hashKey, long timeout)
         *
         * destination: 消息发送到的目的地,不存在时自动创建, 格式为: topicName:tags(主题名:标签名)
         * message: 消息对象
         * hashKey: 使用此键选择消息队列，例如：orderId，productId.....，
         * hashKey 的值可以自定义，只需保证同一顺序批次的消息 hashKey 一致即可，目的是让顺序发送的消息发送到同一个队列，这样在有序消费时保证才能按着顺序消费。
         * timeout；发送超时时间（毫秒）
         */
        try {
            rocketMQTemplate.syncSendOrderly(topic, MessageBuilder.withPayload(dataMap).build(), "orderId");
        } catch (Exception e) {
            //消息发送失败，需要有补偿机制，比如临时保存到数据库中，后期再发送
            e.printStackTrace();
        }
        return "Success! " + new Date();
    }

    /**
     * 异步顺序发送消息：http://localhost:8080/helloFriend/asyncSendOrderly?topic=WMX_TOP:cat
     *
     * @param topic   :主题名称, 如 WMX_TOP, 或者 WMX_TOP:TagA
     * @param dataMap : 消息对象,如 {"id":"787878TYFGH","orderId":"p9009","price":178787.878,"summary":"你好！RocketMQ asyncSendOrderly"}
     *                <p></p>
     *                asyncSendOrderly(String destination, Message<?> message, String hashKey, SendCallback sendCallback)
     *                asyncSendOrderly(String destination, Message<?> message, String hashKey, SendCallback sendCallback,long timeout)
     *                asyncSendOrderly(String destination, Object payload, String hashKey, SendCallback sendCallback)
     *                asyncSendOrderly(String destination, Object payload, String hashKey, SendCallback sendCallback,long timeout)
     *                destination: 消息发送到的目的地,不存在时自动创建, 格式为: topicName:tags(主题名:标签名)
     *                payload: 待发送的消息对象,会自动序列化{@link MessageConverter}为 {@link Message} 进行发送, 无论是普通的 String,或者 POJO 对象都可以
     *                sendCallback: 消息发送结果回调, 发送成功或者失败进入不同的方法
     *                message：消息对象
     *                timeout；发送超时时间（毫秒）
     *                hashKey: 使用此键选择消息队列，例如：orderId，productId.....，这个值可以自定义，目的是让消息发送到同一个队列，这样在有序消费时保证消费顺序。
     *                payload 转 message： Message<?> message = MessageBuilder.withPayload(payload).build()
     */
    @PostMapping("helloFriend/asyncSendOrderly")
    public String asyncSendOrderly(@RequestParam String topic, @RequestBody Map<String, Object> dataMap) {

        rocketMQTemplate.asyncSendOrderly(topic, MessageBuilder.withPayload(dataMap).build(), "hashKey", new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                //成功时，返回状态：SEND_OK
                System.out.println("异步顺序发送消息成功：" + sendResult.getSendStatus());
            }

            @Override
            public void onException(Throwable e) {
                System.out.println("异步顺序发送消息失败：" + e);
                //如果后期需要再重发，则应该进行保存发送失败的消息
            }
        }, 1 * 60 * 1000);
        return "Success! " + new Date();
    }

    /**
     * 事务消息
     * http://localhost:8080/helloFriend/sendMessageInTransaction?topic=wmx2:cat
     *
     * @param topic   :主题名称, 如 WMX_TOP, 或者 WMX_TOP:TagA
     * @param dataMap : 消息对象,如 {"id":"787878YH","orderId":"p9009","price":178787.878,"summary":"你好！RocketMQ asyncSendOrderly"}
     * @return
     */
    @PostMapping("helloFriend/sendMessageInTransaction")
    public String sendMessageInTransaction(@RequestParam String topic, @RequestBody Map<String, Object> dataMap) {
        /**
         *TransactionSendResult sendMessageInTransaction(final String destination,final Message<?> message, final Object arg)
         * destination: 消息发送到的目的地,不存在时自动创建, 格式为: topicName:tags(主题名:标签名)
         * message: 消息对象
         * arg：参数与本地事务执行器一起使用
         */
        try {
            //发送事务消息
            rocketMQTemplate.sendMessageInTransaction(topic, MessageBuilder.withPayload(dataMap).build(), null);
        } catch (Exception e) {
            //消息发送失败，需要有补偿机制，比如临时保存到数据库中，后期再发送
            e.printStackTrace();
        }
        return "Success! " + new Date();
    }


}
